package com.example.hbase;

import com.alibaba.fastjson.JSON;
import com.example.hbase.config.HBaseConfig;
import com.example.hbase.model.Product;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Map;
import java.util.TreeMap;

@Slf4j
public class HBaseServletAJAX extends HttpServlet {
    private static final String TABLE_NAME = "products";
    private static final String COLUMN_FAMILY = "info";

    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        Connection connection = null;
        Table table = null;
        TreeMap<Float, Product> productMap = new TreeMap<>(new java.util.Comparator<Float>() {
            public int compare(Float r1, Float r2) {
                return r2.compareTo(r1);
            }
        });

        response.setContentType("application/json");
        response.setCharacterEncoding("UTF-8");
        PrintWriter out = response.getWriter();

        try {
            connection = HBaseConfig.getConnection();
            table = connection.getTable(TableName.valueOf(TABLE_NAME));
            SingleColumnValueFilter filter = new SingleColumnValueFilter(
                    Bytes.toBytes(COLUMN_FAMILY),
                    Bytes.toBytes("rating"),
                    CompareFilter.CompareOp.GREATER,
                    Bytes.toBytes(0.0f)
            );
            filter.setFilterIfMissing(true);
            Scan scan = new Scan();
            scan.addFamily(Bytes.toBytes(COLUMN_FAMILY));
            scan.setFilter(filter);
            ResultScanner scanner = table.getScanner(scan);

            for (Result result : scanner) {
                String rowKey = Bytes.toString(result.getRow());
                String name = Bytes.toString(result.getValue(Bytes.toBytes(COLUMN_FAMILY), Bytes.toBytes("name")));
                String ratingStr = Bytes.toString(result.getValue(Bytes.toBytes(COLUMN_FAMILY), Bytes.toBytes("rating")));
                float rating;
                try {
                    rating = Float.parseFloat(ratingStr);
                } catch (NumberFormatException e) {
                    log.error("Invalid rating for row " + rowKey + ": " + ratingStr);
                    rating = 0.0f;
                    continue;
                }
                log.info("Row: " + rowKey + ", Name: " + name + ", Rating: " + rating);
                productMap.put(rating, new Product(rowKey, name, rating));
            }
            scanner.close();

            Product[] topProducts = new Product[Math.min(5, productMap.size())];
            int index = 0;
            for (Map.Entry<Float, Product> entry : productMap.entrySet()) {
                if (index < 5) {
                    topProducts[index] = entry.getValue();
                    index++;
                } else {
                    break;
                }
            }

            String jsonData = JSON.toJSONString(topProducts);
            log.info("JSON Data: " + jsonData);
            out.print(jsonData);

        } catch (Exception e) {
            log.error("统计信息失败：{}", e.getMessage());
            out.print("[]"); // Return empty array on error
        } finally {
            if (out != null) out.flush();
            if (table != null) table.close();
            if (connection != null) connection.close();
        }
    }
}